package cn.net.dpi.shp.task;

import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.lang.reflect.Method;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * spring boot 异步执行线程池配置类
 *
 * @author fanfan
 * @since 2021-09-10
 */
@Configuration
@EnableAsync
public class AsyncTheadPool {

    @Bean("kafka_shp_executor")
    public TaskExecutor kafkaShpComsumeExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        // 设置核心线程数
        taskExecutor.setCorePoolSize(1);
        int maxPoolSize = Runtime.getRuntime().availableProcessors() * 2;
        taskExecutor.setMaxPoolSize(maxPoolSize);
        // 设置队列容量
        taskExecutor.setQueueCapacity(1000);
        // 设置线程活跃时间（秒）
        taskExecutor.setKeepAliveSeconds(60);
        // 设置拒绝策略,拒绝策略用默认AbortPolicy,不丢弃任务,失败时抛异常
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        // 等待所有任务结束后再关闭线程池
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);

        taskExecutor.setThreadNamePrefix("Async_shp_TP_");

        taskExecutor.initialize();
        return taskExecutor;
    }

    @Bean
    public AsyncUncaughtExceptionHandler asyncExceptionHandler() {
        return new CustomAsyncExceptionHandler();
    }
}


@Slf4j
class CustomAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {
    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        log.error("async method error:{}", method.getName());
        ex.printStackTrace();
    }
}